package kafka;

import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * kafka测试类
 * USER: lintc 【lintiancong@zhuojianchina.com】
 * DATE: 2017-03-01 09:32
 */
public class KafkaTest {

    public int num = 0;

    private long timeOut;

    private boolean run = true;

    KafkaConsumer<String, String> consumer;

    public KafkaTest(long timeOut, String group) {
        this.timeOut = timeOut;
        consumer = KafkaCustomer.getConsumer(group);
    }

    public int stop() {
        this.run = false;
        return num;
    }

    public void getMsg(String topic) {
        consumer.subscribe(Arrays.asList(topic));
        List<JSONObject> list = new ArrayList<>();
        while (run) {
            ConsumerRecords<String, String> records = consumer.poll(timeOut);
            for (ConsumerRecord<String, String> record : records) {
                num ++;
//                System.out.println(num);
            }
        }
        consumer.close();
        System.out.println(timeOut + " : " + num);
    }

    public void read() {
        while (true) {
            System.out.println(run);
            try {
                Thread.sleep(5 * 100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
//        new Thread(() -> {
//            KafkaTest t = new KafkaTest(50);
//            t.getMsg("app_nbpt");
//            try {
//                Thread.sleep(20 * 1000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            t.stop();
//            System.out.println("结束 ： " + t.stop());
//        }).start();
        KafkaTest t = new KafkaTest(100, "app_nbpt");
        KafkaTest t2 = new KafkaTest(50, "app_zyyy");
        try {
            Thread.sleep(10 * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(() -> {
            t.getMsg("app_nbpt");
        }).start();
        new Thread(() -> {
            t2.getMsg("app_nbpt");
        }).start();
        try {
            Thread.sleep(20 * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        t.stop();
        t2.stop();
    }
}
